package com.csw.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo17Checkpoint {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setMaster("local")
      .setAppName("Checkpoint")

    val sc: SparkContext = new SparkContext(conf)

    /**
      * checkpoint：将rdd的数据保存到hdfs中，
      * 1、当第一个job执行完成之后会从最后一个rdd向前回溯，对调用了checkpoint的rdd打上标记
      * 2、另外启动一个job重写计算rdd的数据，并将rdd的数据保存到hdfs
      *
      * 调优： 在checkpoint之前进行cache  可以优化效率
      *
      * 主要在spark  streaming 中使用
      *
      */

    //设置checkpoint的路径
    sc.setCheckpointDir("spark/data/checkpoint")


    val studentRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    val mapstudentRDD: RDD[String] = studentRDD.map(i => {
      println("map---")
      i
    })

    mapstudentRDD.cache()

    //设置checkpoint的rdd
    mapstudentRDD.checkpoint()


    //统计班级的人数
    val clazzNumRDD: RDD[(String, Int)] = mapstudentRDD.map(i => (i.split(",")(4), 1))
    val clazzCountRDD: RDD[(String, Int)] = clazzNumRDD.reduceByKey(_ + _)

    clazzCountRDD.foreach(println)
    //统计性别的人数
    val genderNumRDD: RDD[(String, Int)] = mapstudentRDD.map(i => (i.split(",")(3), 1))
    val genderCountRDD: RDD[(String, Int)] = genderNumRDD.reduceByKey((x, y) => x + y)

    genderCountRDD.foreach(println)

    //统计年龄的人数
    val ageNumRDD: RDD[(String, Int)] = mapstudentRDD.map(i => (i.split(",")(2), 1))
    val ageCountRDD: RDD[(String, Int)] = ageNumRDD.reduceByKey(_ + _)

    ageCountRDD.foreach(println)

  }
}
